
#include "stdafx.h"
#include "UdpCallBenchApp.h"
#include "UdpCallBenchMgr.h"
#include "ReloadCode.h"
#include "ReloadPacket.h"
#include "ReloadConfigure.h"
#include "ReloadUdpCallPacket.h"

UdpCallBenchMgr::UdpCallBenchMgr( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, std::size_t _workerPoolSize /*= 1*/ )
:asyframe::ActorMgrBase(_workerPoolSize),pRemoteEndpoint_(_pRemoteEndpoint)
{

}

UdpCallBenchMgr::~UdpCallBenchMgr()
{
	pRemoteEndpoint_.reset();
}

retT UdpCallBenchMgr::start()
{
	//boost::shared_ptr<UdpCallBenchApp> pUdpCallBenchApp = boost::dynamic_pointer_cast<UdpCallBenchApp>(asyframe::getApp());
	//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::setTestCodeSize, pUdpCallBenchApp, 20)));
	for(int i=0; i < 20; i++)
	{
		boost::shared_ptr<asyframe::ActorBase> pActor(new UdpCallBenchActor(i));
		R(makeActor(pActor));
		boost::dynamic_pointer_cast<UdpCallBenchActor>(pActor)->setRemoteEndpoint(pRemoteEndpoint_);
	}
	R(asyframe::ActorMgrBase::start());
	RETSUCCESS;
}

//////////////////////////////////////////////////////////////////

UdpCallBenchActor::UdpCallBenchActor(int _index)
:pUdpHandler_(),pRemoteEndpoint_(),seed_(0),success_(0),fail_(0),timeout_(0),error_(0),index_(_index)
{

}

UdpCallBenchActor::~UdpCallBenchActor()
{
	pUdpHandler_.reset();
	pRemoteEndpoint_.reset();
	seed_ = 0;
	success_ = 0;
	fail_ = 0;
	timeout_ = 0;
	error_ = 0;
	index_ = -1;
}

retT UdpCallBenchActor::init()
{
	R(asyframe::ActorBase::init());
	pUdpHandler_.reset(new asyframe::UdpHandler());
	R(pUdpHandler_->attachHandlerOwner(shared_from_this()));
	seed_ = rand();
	RETSUCCESS;
}

retT UdpCallBenchActor::start()
{
	R(asyframe::ActorBase::start());
	R(sendOneRequest(true));
	RETSUCCESS;
}

retT UdpCallBenchActor::close()
{
	R(asyframe::ActorBase::close());
	boost::shared_ptr<UdpCallBenchApp> pUdpCallBenchApp = boost::dynamic_pointer_cast<UdpCallBenchApp>(asyframe::getApp());
	R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::gatherResult, pUdpCallBenchApp, success_, fail_, timeout_, error_)));
	RETSUCCESS;
}

void UdpCallBenchActor::setRemoteEndpoint( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
{
	pRemoteEndpoint_ = _pRemoteEndpoint;
}

retT UdpCallBenchActor::sendOneRequest(bool isInit)
{
	std::string resourceKey = "test";
	resourceKey += seed_;
	resourceKey += (1000 + (success_ + fail_ + timeout_ + error_)%1000);
	resourceKey += "@open-ims.test";
	SendStoreReqUdpCallPacket sendStoreReqUdpCallPacket;
	sendStoreReqUdpCallPacket.overlayInstanceName_.setValue(((UdpCallBenchAppConfigure*)asyframe::getApp()->getConfigure())->instanceName_);
	sendStoreReqUdpCallPacket.resourceName_.setValue(resourceKey);

	StoreKindData storeKindData;
	storeKindData.init(KINDID_GENERIC_SINGLE);
	storeKindData.setSingleValueEntry("127.0.0.1");
	sendStoreReqUdpCallPacket.kindData_.array_.push_back(storeKindData);

	if(isInit)
	{
		pUdpHandler_->getSocket()->open(pRemoteEndpoint_->protocol());
	}

	if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
	{
		boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
		pUdpHandlerContext->bytesToTransfer_ = sendStoreReqUdpCallPacket.size();
		boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
		R(sendStoreReqUdpCallPacket.writeToBuffer(pBuffer));
		R(pUdpHandler_->asyncWriteTo(pBuffer, pRemoteEndpoint_, pUdpHandlerContext));
	}
	else
		RET(ERROR_COMMON_SHARED_PTR_NULL);

	if(isInit)
	{
		boost::shared_ptr<boost::asio::ip::udp::endpoint> pLocalEndpoint(new boost::asio::ip::udp::endpoint(
			pUdpHandler_->getSocket()->local_endpoint()));
		pUdpHandler_->setLocalEndpoint(pLocalEndpoint);
		R(pUdpHandler_->reset());
	}

	if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
	{
		boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
		boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, maxReloadUdpCallPacketSize));
		pUdpHandlerContext->pTimerDuration_.reset(new boost::posix_time::seconds(UDPCALLBENCH_WAIT_RESPONSE_SECONDS));
		pUdpHandlerContext->bytesToTransfer_ = maxReloadUdpCallPacketSize;
		pUdpHandlerContext->pRemoteEndpoint_.reset(new boost::asio::ip::udp::endpoint());
		R(pUdpHandler_->asyncReadFrom(pBuffer, pUdpHandlerContext->pRemoteEndpoint_, pUdpHandlerContext,
			boost::bind(&UdpCallBenchActor::onReadStoreAnsResponse, boost::dynamic_pointer_cast<UdpCallBenchActor>(shared_from_this()),
			resourceKey, pUdpHandlerContext, _1, _2, _3)));
	}
	else
		RET(ERROR_COMMON_SHARED_PTR_NULL);
	RETSUCCESS;
}

retT UdpCallBenchActor::onReadStoreAnsResponse( std::string _resourceKey, boost::shared_ptr<asyframe::UdpHandlerContext> _pUdpHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
{
	R(asyframe::IUdpAction::onReadFrom(_pUdpHandlerContext, _bytesTransferred, e, _timeoutCount));
	R(sendOneRequest());
	boost::shared_ptr<UdpCallBenchApp> pUdpCallBenchApp = boost::dynamic_pointer_cast<UdpCallBenchApp>(asyframe::getApp());
	if(!e)
	{
		if(_pUdpHandlerContext && _pUdpHandlerContext->pRemoteEndpoint_)
		{
			R(_pUdpHandlerContext->reset());
			if(_timeoutCount == 0)
			{
				if(_pUdpHandlerContext->pBuffer_)
				{
					ReloadUdpCallResultBase reloadUdpCallResultBase;
					R(reloadUdpCallResultBase.readFromBuffer(_pUdpHandlerContext->pBuffer_));
					if(!reloadUdpCallResultBase.result_)
					{
						++success_;
						//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incSuccess, pUdpCallBenchApp, index_)));

						//SendFetchReqUdpCallPacket sendFetchReqUdpCallPacket;
						//sendFetchReqUdpCallPacket.overlayInstanceName_.setValue(((UdpCallBenchAppConfigure*)asyframe::getApp()->getConfigure())->instanceName_);
						//sendFetchReqUdpCallPacket.resourceName_.setValue(_resourceKey);

						//StoredDataSpecifier specifier;
						//specifier.init(KINDID_GENERIC_SINGLE);
						//sendFetchReqUdpCallPacket.specifiers_.array_.push_back(specifier);

						//boost::shared_ptr<boost::asio::ip::udp::endpoint> pRemoteEndpoint(new boost::asio::ip::udp::endpoint(
						//	boost::asio::ip::address::from_string("127.0.0.1"),ReloadUdpCall_ListeningPort_Fetch));

						//if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
						//{
						//	boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
						//	pUdpHandlerContext->bytesToTransfer_ = sendFetchReqUdpCallPacket.size();
						//	boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
						//	R(sendFetchReqUdpCallPacket.writeToBuffer(pBuffer));
						//	R(pUdpHandler_->asyncWriteTo(pBuffer, pRemoteEndpoint, pUdpHandlerContext));
						//}
						//else
						//	RET(ERROR_COMMON_SHARED_PTR_NULL);

						//if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
						//{
						//	boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
						//	boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, maxReloadUdpCallPacketSize));
						//	pUdpHandlerContext->pTimerDuration_.reset(new boost::posix_time::seconds(UDPCALLBENCH_WAIT_RESPONSE_SECONDS));
						//	pUdpHandlerContext->bytesToTransfer_ = maxReloadUdpCallPacketSize;
						//	pUdpHandlerContext->pRemoteEndpoint_.reset(new boost::asio::ip::udp::endpoint());
						//	R(pUdpHandler_->asyncReadFrom(pBuffer, pUdpHandlerContext->pRemoteEndpoint_, pUdpHandlerContext,
						//		boost::bind(&UdpCallBenchActor::onReadFetchAnsResponse, boost::dynamic_pointer_cast<UdpCallBenchActor>(shared_from_this()),
						//		pUdpHandlerContext, _1, _2, _3)));
						//}
						//else
						//	RET(ERROR_COMMON_SHARED_PTR_NULL);
					}
					else
					{
						++fail_;
						//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incFail, pUdpCallBenchApp, index_)));
					}
				}
			}
			else
			{
				++timeout_;
				//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incTimeout, pUdpCallBenchApp, index_)));
			}
		}
	}
	else
	{
		LOG_INFO(boost::format( "errorNo %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		++error_;
		//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incError, pUdpCallBenchApp, index_)));
	}
	RETSUCCESS;
}

retT UdpCallBenchActor::onReadFetchAnsResponse( boost::shared_ptr<asyframe::UdpHandlerContext> _pUdpHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
{
	R(asyframe::IUdpAction::onReadFrom(_pUdpHandlerContext, _bytesTransferred, e, _timeoutCount));
	R(sendOneRequest());
	boost::shared_ptr<UdpCallBenchApp> pUdpCallBenchApp = boost::dynamic_pointer_cast<UdpCallBenchApp>(asyframe::getApp());
	if(!e)
	{
		if(_pUdpHandlerContext && _pUdpHandlerContext->pRemoteEndpoint_)
		{
			R(_pUdpHandlerContext->reset());
			if(_timeoutCount == 0)
			{
				if(_pUdpHandlerContext->pBuffer_)
				{
					SendFetchReqUdpCallResult sendFetchReqUdpCallResult;
					R(sendFetchReqUdpCallResult.readFromBuffer(_pUdpHandlerContext->pBuffer_));
					if(!sendFetchReqUdpCallResult.result_)
					{
						++success_;
						//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incSuccess, pUdpCallBenchApp, index_)));
						//std::cout << sendFetchReqUdpCallResult.kindResponses_.array_.size() << std::endl;
					}
					else
					{
						++fail_;
						//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incFail, pUdpCallBenchApp, index_)));
					}
				}
			}
			else
			{
				++timeout_;
				//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incTimeout, pUdpCallBenchApp, index_)));
			}
		}
	}
	else
	{
		LOG_INFO(boost::format( "errorNo %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		++error_;
		//R(pUdpCallBenchApp->post(boost::bind(&UdpCallBenchApp::incError, pUdpCallBenchApp, index_)));
	}
	RETSUCCESS;
}
